package com.atguigu.util;

import com.atguigu.constant.CommonConstantValue;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.log4j.Logger;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;


/**
 * @ClassName HdfsUtil
 * @DeScription HdfsUtil
 * @Author 剧情再美终是戏
 * @Date 2019/11/13 18:56
 * @Version 1.0
 **/
public class HdfsUtil {

    private static Logger logger = Logger.getLogger(HdfsUtil.class);

    public static final String HDFS_FILE_SYSTEM_HOST = CommonConstantValue.HDFS_CONNECT_ADRESS;

    private HdfsUtil() {
    }


    /**
     * 获取一个文件系统实例
     *
     * @param fileSystemHost 文件系统的主机名，为空的情况下就是本地文件系统
     * @return org.apache.hadoop.fs.FileSystem 文件系统实例
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static FileSystem getFileSystem(String fileSystemHost) {
        FileSystem fs = null;
        Configuration conf = new Configuration();
        try {
            if (StringUtils.isNotBlank(fileSystemHost)) {
                fs = FileSystem.get(new URI(fileSystemHost), conf);
            } else {
                fs = FileSystem.get(conf);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        return fs;
    }

    /**
     * 获取一个文件系统实例
     *
     * @param conf 配置文件
     * @return org.apache.hadoop.fs.FileSystem
     * @Author 剧情再美终是戏
     * @Date 2019/11/18
     * @Version 1.0
     **/
    public static FileSystem getFileSystem(Configuration conf) {
        FileSystem fs = null;
        try {
            fs = FileSystem.get(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return fs;
    }

    /**
     * 关闭多个文件系统
     *
     * @param fs 需要操作的文件系统实例
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void closeStream(Closeable... fs) {
        if (null == fs) {
            return;
        }
        for (Closeable f : fs) {
            if (null != f) {
                IOUtils.closeStream(f);
                logger.debug(f.getClass().getName() + "已关闭");
            }
        }
    }

    /**
     * 创建文件夹
     *
     * @param fs   需要操作的文件系统实例
     * @param path 创建文件夹的路径
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void mkdir(FileSystem fs, String path) {
        if (null == fs || StringUtils.isBlank(path)) {
            logger.error("参数有误，fs, path是必须的!");
            return;
        }

        try {
            boolean mkdirs = fs.mkdirs(new Path(path));
            logger.debug(String.format("创建文件夹路径为%s,创建%b", path, mkdirs));
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            closeStream(fs);
        }
    }

    /**
     * 重命名文件夹
     *
     * @param fs      需要操作的文件系统实例
     * @param oldPath 重命名的源文件路径
     * @param newPath 重命名的目标路径
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void renameDir(FileSystem fs, String oldPath, String newPath) {
        if (null == fs || StringUtils.isBlank(oldPath) || StringUtils.isBlank(newPath)) {
            logger.error("参数有误，创建文件夹时fs, oldPath, newPath!");
            return;
        }

        try {
            boolean rename = fs.rename(new Path(oldPath), new Path(newPath));
            logger.debug(String.format("文件夹重命名，原名称%s，修改为%s, 重命名%b", oldPath, newPath, rename));
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            closeStream(fs);
        }
    }

    /**
     * 删除文件
     *
     * @param fs        需要操作的文件系统实例
     * @param path      删除文件的路径
     * @param recursion 是否递归删除
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void delete(FileSystem fs, String path, boolean recursion) throws IOException {
        if (null == fs || StringUtils.isBlank(path)) {
            logger.error("参数有误，fs, path是必须的!");
            return;
        }
        delete(fs, new Path(path), recursion);
    }

    /**
     * 删除文件
     *
     * @param fs        需要操作的文件系统实例
     * @param path      删除文件的路径
     * @param recursion 是否递归删除
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/20
     * @Version 1.0
     **/
    public static void delete(FileSystem fs, Path path, boolean recursion) throws IOException {
        if (null == fs || null == path) {
            logger.error("参数有误，fs, path是必须的!");
            return;
        }
        if (!fs.exists(path)) {
            return;
        }
        try {
            boolean delete = fs.delete(path, recursion);
            logger.debug(String.format("删除文件，文件路径是%s，删除%b", path, delete));
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            closeStream(fs);
        }
    }

    /**
     * 文件上传
     *
     * @param fs        需要操作的文件系统实例
     * @param delSrc    是否删除源文件，如果为true就是剪切
     * @param overwrite 是否覆盖文件
     * @param src       源文件路径
     * @param dst       目标文件路径
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void upload(FileSystem fs, boolean delSrc, boolean overwrite, String src, String dst) {
        if (StringUtils.isBlank(src) || StringUtils.isBlank(dst)) {
            logger.error("上传文件时，src， dst参数是必须的!");
            return;
        }
        try {
            fs.copyFromLocalFile(delSrc, overwrite, new Path(src), new Path(dst));
            logger.debug(String.format("上传文件成功，源文件路径%s， 目标路径%s", src, dst));
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(String.format("上传文件%b，失败原因%s", true, e));
        } finally {
            // 关闭资源
            closeStream(fs);
        }
    }


    /**
     * 文件下载
     *
     * @param fs                    需要操作的文件系统实例
     * @param delSrc                是否删除源文件
     * @param src                   下载路径
     * @param dst                   源文件路径
     * @param useRawLocalFileSystem 是否使用原始文件系统，就是是否需要下载检验和文件
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void download(FileSystem fs, boolean delSrc, String src, String dst, boolean useRawLocalFileSystem) {
        if (StringUtils.isBlank(src) || StringUtils.isBlank(dst)) {
            logger.error("文件下载时，src， dst参数是必须的!");
            return;
        }
        try {
            fs.copyToLocalFile(delSrc, new Path(dst), new Path(src), useRawLocalFileSystem);
            logger.debug(String.format("文件下载%b，下载路径%s， 目标路径%s", true, dst, src));
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(String.format("文件下载失败，失败原因%s", e));
        } finally {
            // 关闭资源
            closeStream(fs);
        }
    }

    /**
     * 文件详情查看
     *
     * @param fs        需要操作的文件系统实例
     * @param path      文件路径
     * @param recursion 是否递归
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void listFiles(FileSystem fs, String path, boolean recursion) {
        if (null == fs || StringUtils.isBlank(path)) {
            logger.error("参数有误，fs, path是必须的!");
            return;
        }
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = null;
        try {
            locatedFileStatusRemoteIterator = fs.listFiles(new Path(path), recursion);
            while (locatedFileStatusRemoteIterator.hasNext()) {
                LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
                logger.debug(String.format("文件名是%s, 是否文件夹%b", next.getPath().getName(), next.isDirectory()));
                BlockLocation[] blockLocations = next.getBlockLocations();
                if (null != blockLocations) {
                    int i = 1;
                    for (BlockLocation blockLocation : blockLocations) {
                        logger.debug(String.format("-----------第%d块,详细信息如下：-------------", i));
                        logger.debug(blockLocation);
                        i++;
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 从本地自定义上传指定长度(k)的文件到hdfs
     *
     * @param local 本地文件系统
     * @param fs    hdfs文件系统
     * @param src   本地文件路径
     * @param dst   hdfs文件路径
     * @param k     上传的大小
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/13
     * @Version 1.0
     **/
    public static void diyUpload(FileSystem local, FileSystem fs, String src, String dst, long k) {
        try {
            FSDataInputStream in = local.open(new Path(src));
            FSDataOutputStream out = fs.create(new Path(dst));

            byte[] bb = new byte[1024];
            for (int i = 0; i < k; i++) {
                in.read(bb);
                out.write(bb);
            }
            logger.debug(String.format("上传文件成功，源文件路径%s， 目标路径%s", src, dst));
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(String.format("上传文件%b，失败原因%s", true, e));
        } finally {
            closeStream(fs, local);
        }
    }

    // 自定义下载指定长度的文件
    public static void diyDownload(FileSystem local, FileSystem fs, String src, String dst) {
        try {
//            FileOutputStream out = new FileOutputStream(new File(src));
//            FSDataOutputStream out = local.create(new Path(src));
//        FileOutputStream out = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part1"));


            FSDataOutputStream out = local.create(new Path("e:/hadoop-2.7.2.tar.gz.part11"));

            FSDataInputStream in = fs.open(new Path(dst));

            byte[] bb = new byte[4096];
            for (int i = 0; i < 1024 * 128 / 4; i++) {
                in.read(bb);
                out.write(bb);
            }
            closeStream(out, in);
            logger.debug("---------------ok-------------");
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(String.format("文件下载失败，失败原因%s", e));
        } finally {
            // 关闭资源
            closeStream(fs, local);
        }
    }

    /**
     * 压缩类
     *
     * @param sourcePath 需要压缩的文件路径
     * @param codec      压缩格式
     * @param conf       配置
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/20
     * @Version 1.0
     **/
    public static void compress(String sourcePath, String codec, Configuration conf) throws Exception {
        // 根据压缩格式获取一个输出流
        Class<?> aClass = Class.forName(codec);
        CompressionCodec compressionCodec = (CompressionCodec) ReflectionUtils.newInstance(aClass, conf);
        String outPath = sourcePath + compressionCodec.getDefaultExtension();
        CompressionOutputStream out = compressionCodec.createOutputStream(new FileOutputStream(new File(outPath)));

        // 获取文件的输入流
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path(sourcePath);
        if (fs.exists(path)) {
            // 如果文件已经存在，就删除
            fs.delete(path, true);
        }
        FSDataInputStream in = fs.open(path);

        // 输入和输出流的拷贝
        IOUtils.copyBytes(in, out, 4096, true);

        // 关闭文件系统
        if (null != fs) {
            fs.close();
        }
        logger.debug("压缩完成~文件为 : " + outPath);
    }

    /**
     * 解压缩
     *
     * @param sourcePath 输入文件的路径（压缩包的路径）
     * @param outPath    输出文件的路径
     * @param conf       配置
     * @return void
     * @Author 剧情再美终是戏
     * @Date 2019/11/20
     * @Version 1.0
     **/
    public static void decompression(String sourcePath, String outPath, Configuration conf) throws Exception {
        // 根据压缩文件获取压缩的输入流
        CompressionCodec compressionCodec = new CompressionCodecFactory(conf).getCodec(new Path(sourcePath));
        CompressionInputStream in = compressionCodec.createInputStream(new FileInputStream(new File(sourcePath)));

        // 获取输出文件的输出流
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path(outPath);
        if (fs.exists(path)) {
            // 如果文件已经存在，就删除
            fs.delete(path, true);
        }
        FSDataOutputStream out = fs.create(path);

        // 输入和输出流的拷贝
        IOUtils.copyBytes(in, out, 4096, true);

        // 关闭文件系统
        if (null != fs) {
            fs.close();
        }
        logger.debug("解压缩完成~文件为 : " + outPath);
    }
}
